package com.emrubik.sparkstream2;

import java.io.Serializable;
import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.TaskContext;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka010.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

/**
 * Hello world!
 */
public class App implements Serializable {

    /**
     * 日志輸出
     */
    protected final Logger log = LoggerFactory.getLogger(App.class);

    /**
     * StreamingContext 对象
     */
    private transient JavaStreamingContext streamingContext = null;

    private void start(){
        System.out.println("Hello World!");
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "10.10.30.213:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);

        Collection<String> topics = Arrays.asList("stream-test");

        SparkConf sparkConf = new SparkConf().setAppName("Stream2").setMaster("local[4]");

        streamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(3));

        JavaInputDStream<ConsumerRecord<String, String>> stream =
                KafkaUtils.createDirectStream(
                        streamingContext,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
                );


        JavaPairDStream<String,String> kafkamsg =stream.mapToPair(record -> new Tuple2<>(record.key(), record.value()));

        kafkamsg.foreachRDD(new VoidFunction<JavaPairRDD<String, String>>() {
            @Override
            public void call(JavaPairRDD<String, String> stringStringJavaPairRDD) throws Exception {
                stringStringJavaPairRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, String>>>() {
                    @Override
                    public void call(Iterator<Tuple2<String, String>> tuple2Iterator) throws Exception {
                        if(tuple2Iterator.hasNext()){
                            log.info("接收到消息： " + tuple2Iterator.next()._2);
                        }
                    }
                });
            }
        });


        streamingContext.start();
    }

    public static void main(String[] args) {
        try {
            App app = new App();
            app.start();
            app.streamingContext.awaitTermination();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
